package at.ac.tuwien.swa.sparrow.peer

import ac.at.tuwien.infosys.swa.audio.Fingerprint
import akka.actor.SupervisorStrategy.Stop
import akka.actor._
import at.ac.tuwien.swa.sparrow._
import at.ac.tuwien.swa.sparrow.common.AuthToken
import at.ac.tuwien.swa.sparrow.common.PeerInfo
import at.ac.tuwien.swa.sparrow.common.Ticket
import at.ac.tuwien.swa.sparrow.common.TrackInfo
import at.ac.tuwien.swa.sparrow.peer.PeerCommand._
import at.ac.tuwien.swa.sparrow.peer.component.{RequestHandler, IndexingWorker, MusicLibrary}
import at.ac.tuwien.swa.sparrow.peer.conf.SpringConfig
import at.ac.tuwien.swa.sparrow.server.PeerService
import at.ac.tuwien.swa.sparrow.server.security.AuthenticatingHttpInvokerRequestExecutor
import com.google.common.io.Files
import com.google.common.net.HostAndPort
import com.typesafe.config.{ConfigValueFactory, ConfigFactory, ConfigRenderOptions, Config}
import com.typesafe.scalalogging.log4j.Logging
import java.net.NetworkInterface
import java.nio.charset.Charset
import java.util.UUID
import javax.inject.Inject
import org.springframework.context.annotation.AnnotationConfigApplicationContext
import org.springframework.remoting.httpinvoker.HttpInvokerProxyFactoryBean
import org.springframework.stereotype.Component
import scala.Some
import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Try, Random}

object PeerActor extends App {

	val ctx = new AnnotationConfigApplicationContext(classOf[SpringConfig])
	@Inject var requestHandler: RequestHandler = _
	@Inject var peer: ActorRef = _
	@Inject var config: Config = _
	@Inject var indexingWorker: IndexingWorker = _

	ctx.getAutowireCapableBeanFactory.autowireBean(this)
	global.execute(new Runnable {
		def run() = config getStringList "library.paths" foreach indexingWorker.scan
	})

	readLine("Press ENTER to exit\n")
	println("Stopping peer...")
	if (peer != null) peer ! Stop
	ctx.close()
}

@Component
class PeerActorFactory @Inject()(system: ActorSystem, peerService: PeerService, musicLibrary: MusicLibrary)
	extends Logging {

	var actorType: Class[_ <: Actor] = classOf[PeerActor]
	@Inject var config: Config = _

	def getObjectType = classOf[PeerActor]

	def createInstance = {
		system.actorOf(Props(actorType, peerService, musicLibrary, config))
	}
}

class PeerActor @Inject()(peerService: PeerService, library: MusicLibrary, var config: Config)
	extends Actor with Logging {

	val serverNotifier = context.actorOf(Props(classOf[ServerNotifier], this))
	val peers = mutable.Map[PeerInfo, Option[PeerInterface]]() withDefault connect

	lazy val address = HostAndPort fromParts((NetworkInterface.getNetworkInterfaces.toList find (!_.isLoopback)).get.getInterfaceAddresses.head.getAddress.getHostAddress, config getInt "port")
	var peerInfo: PeerInfo = _
	var peerId: UUID = _

	override def preStart() {
		peerInfo = authenticate()
		context.system.scheduler.schedule(PEER_TIMEOUT, PEER_TIMEOUT) {
			serverNotifier ! PeerNetwork(peers.keySet + peerInfo.copy(lastOnline = System.currentTimeMillis))
		}

		sys.addShutdownHook {
			postStop()
		}
		context.system.registerOnTermination({
			postStop()
		})
	}

	override def postStop() {
		if (peerId != null) {
			peerService.logout(peerId)
			peerId = null
		}
	}

	def authenticate(): PeerInfo = {
		if (config.hasPath("peerId")) {
			peerId = UUID fromString (config getString "peerId")
			AuthenticatingHttpInvokerRequestExecutor.properties = Map("peerId" -> peerId)
		} else {
			val username = config getString "username"
			val password = config getString "password"
			logger.info("Registering as {}", username)
			AuthenticatingHttpInvokerRequestExecutor.properties = Map("username" -> username, "password" -> password)
			peerId = peerService.register(username, password)
			AuthenticatingHttpInvokerRequestExecutor.properties = Map("peerId" -> peerId)

			config = config withValue("peerId", ConfigValueFactory fromAnyRef peerId.toString)
			saveConfiguration()
		}
		logger.info("Authenticating as {}", peerId)
		peerService.login(peerId, address)
	}

	def saveConfiguration() {
		val map = config.root.unwrapped map (e => "swazam." + e._1 -> e._2)
		val string = (ConfigFactory parseMap map).root render ConfigRenderOptions.defaults.setOriginComments(false)
		Files write(string, CONFIG_FILE_PEER, Charset.defaultCharset())
	}

	def receive: Actor.Receive = {
		case SongRecognition(t) if t.isExpired => logger.info(s"Got expired ticket: ${t.id}")
		case req@SongRecognition(t) if t.peers contains address => handleForward(req)
		case req: SongRecognition => handleRequest(req)
		case pi: PeerIdentify => sender ! PeerIdentity(peerInfo)
		case pl: PeerLookup => sender ! PeerNetwork(peers.keySet + peerInfo)
		case PeerNetwork(ps: Set[PeerInfo]) => mergePeers(ps)
		case Notification(m) => serverNotifier ! m
	}

	def mergePeers(ps: Set[PeerInfo]) {
		for (p <- ps) {
			val option = peers.find(_._1.id == p.id)
			option match {
				case None => peers += (p -> null)
				case Some(oldP) if oldP._1.lastOnline < p.lastOnline => {
					peers -= oldP._1
					peers += (p -> oldP._2)
				}
			}
		}
		peers -= peerInfo
	}

	def handleRequest(req: SongRecognition) {
		logger.info(s"Processing request ${req.ticket.id} (${req.ticket.hopCount}) of ${req.ticket.userId}")
		req.ticket.peers += this.address
		val result = library(req.ticket.fingerPrint) map (i => Result(req.ticket, Some(i._2)))
		logger.info(s"Request ${req.ticket.id} is $result")
		if (result.isDefined) {
			serverNotifier ! result.get
		} else {
			val fwd = handleForward(req)
			if (fwd.isDefined) req.ticket.peers += fwd.get.address
			serverNotifier ! Result(req.ticket, None)
		}
	}

	def selectPeers(ticket: Ticket) = {
		Random shuffle (peers filterKeys (!ticket.peers.contains(_)))
	}

	def connect(peerInfo: PeerInfo): Some[PeerInterface] = {
		val address = peerInfo.address
		val factoryBean = new HttpInvokerProxyFactoryBean
		factoryBean.setServiceUrl(s"http://${address.getHostText}:${address.getPortOrDefault(DEFAULT_PEER_PORT)}/remoting/PeerInterface")
		factoryBean.setServiceInterface(classOf[PeerInterface])
		factoryBean.afterPropertiesSet()
		Some(factoryBean.getObject.asInstanceOf[PeerInterface])
	}

	def handleForward(req: SongRecognition): Option[PeerInfo] = {

		def doForward(candidates: Set[PeerInfo], excludes: Set[HostAndPort]): Option[PeerInfo] = {
			if (candidates.isEmpty) return None
			val receiver = candidates filterNot (c => excludes contains c.address) find {
				case c => Try {
					val peerInterface: PeerInterface = peers(c) match {
						case None => peers.update(c, connect(c)); peers(c).get
						case Some(pi) => pi
					}
					peerInterface request req.ticket
				}.isSuccess
			}
			if (receiver.isEmpty) {
				val newPeers = peerService lookup(req.ticket.peers ++: (candidates map (_.address)), 10) map (_ -> null)
				peers ++: newPeers
				return doForward(newPeers map (_._1), excludes)
			}
			receiver
		}

		if (peers.isEmpty) peers ++= (peerService lookup (Set(peerInfo.address), 10) map (_ -> None))
		val candidates = (selectPeers(req.ticket) map (_._1)).toSet
		doForward(candidates, req.ticket.peers.toSet)
	}

	class ServerNotifier extends Actor with ActorLogging {
		def receive: Actor.Receive = {
			case Result(t, i) =>
				logger.info(s"Updating ticket information for ${t.id} with $i")
				peerService.update(t, i)
			case PeerNetwork(ps) =>
				logger.info(s"Updating peer information: $ps")
				peerService.update(ps.toSet)
			case x => logger.warn(s"CANNOT HANDLE MESSAGE: $x")
		}
	}

}

object PeerCommand {

	abstract class PeerIdentify

	case object PeerIdentify extends PeerIdentify

	case class PeerIdentity(peerInfo: PeerInfo)

	case class ClientRequest(authToken: AuthToken, fp: Fingerprint)

	case class SongRecognition(ticket: Ticket)

	case class Result(ticket: Ticket, info: Option[TrackInfo])

	abstract class PeerLookup

	case object PeerLookup extends PeerLookup

	case class PeerNetwork(peers: scala.collection.Set[PeerInfo])

	case class Notification(msg: AnyRef)
}
